<?php
/**
 * 服务：推送数据
 * 从redis队列中取出消息并post
 * date：2018-9-7
 */

define('APP_ROOT', __DIR__);

$continueAsk = true;

$config = include APP_ROOT . '/../config/config.php';
define('URL', 'http://wycapi.gzjt.gov.cn/api/app/common/binapi');

echo '======[Guangzhou]Post Server Start====', PHP_EOL;

$redis = new Redis();
$redis->connect($config['redis_host'], $config['redis_port']) or die('Redis connect failed!');
if (isset($config['redis_pass'])) {
    $redis->auth($config['redis_pass']) or die('Redis auth failed!');
}
if (isset($config['redis_db'])) {
    $redis->select($config['redis_db']) or die('Redis select db failed!');
}

$pid = getmypid();
echo 'Redis Connect! pid:', $pid;

if ($redis->incr('cqiu_test_' . $pid)) {
    $redis->decr('cqiu_test_' . $pid);
} else {
    die('Redis write failed!');
}

//================
//reset error data for re-push
if (isset($argv[1])) {
    //reset temp queue
    $tmp_queues = $redis->keys('tmp_queue:*');
    echo 'reset tmp_queue', PHP_EOL;
    print_r($tmp_queues);
    $ids = trim(fgets(STDIN));
    foreach (preg_split('#\s+#', $ids) as $id) {
        while ($redis->rpoplpush('tmp_queue' . $id, 'guangdong:queue')) {
            ;
        }
    }

    //reset error queues
    $tmp_queues = $redis->keys('guangdong:queue:*');
    print_r($tmp_queues);
    echo PHP_EOL, 'reset guangdong:queue:[code] (default all)>';
    $ids = trim(fgets(STDIN));
    if ($ids == '') {
        $ids = &$tmp_queues;
    } else {
        if ($ids[0] == '-') {
            $ids = preg_split('#\s+#', substr($ids, 1));
            foreach ($tmp_queues as $i => $queueName) {
                if (in_array($i, $ids)) {
                    unset($tmp_queues[$i]);
                }
            }
        } else {
            $ids = preg_split('#\s+#', $ids);
            foreach ($tmp_queues as $i => $queueName) {
                if (!in_array($i, $ids)) {
                    unset($tmp_queues[$i]);
                }
            }
        }
        $ids = $tmp_queues;
    }
    foreach ($ids as $id) {
        while ($redis->rpoplpush('guangdong:queue:' . $id, 'guangdong:queue')) {
            ;
        }
    }

    exit('=======done!' . PHP_EOL);
}

//$apis = include APP_ROOT . '/config/net-car-apis.php';

//================
while (1) {

    $msg = $redis->rpoplpush('guangdong:queue', $tmpKey = 'guangdong:tmp_queue:' . $pid);

    if (!$msg) {
        continue;
    }

    _log($msg);

    list($fileName, $content) = explode('|', $msg);
    $content = '{"items":[' . $content . ']}';
    $resultCode = send($fileName, $content);

    _log($resultCode);

//    if ($resultCode == 201) {
//        // success => remove it
//        $redis->lRem($tmpKey, $msg, 1);
//        $redis->incr('guangdong:done');
//    } else {
//        // put down to failed queue
//        if (!is_int($resultCode)) {
//            $resultCode = '000';
//        }
//        $redis->rpoplpush($tmpKey, 'guangdong:queue:' . $resultCode);
//        $redis->incr('guangdong:failed');
//    }


    //----for test!
    if ($continueAsk) {
        if ($continueAsk > 1) {
            sleep($continueAsk);
        } else {
            echo PHP_EOL, str_repeat('_', 50), 'continue? > ';
            $continue = trim(fgets(STDIN));
            if ($continue == 'exit') {
                exit;
            } elseif ($continue == 'yes') {
                $continueAsk = false;
            } elseif (is_int($continue)) {
                $continueAsk = $continue;
            }
        }
    }

}

//===============

/**
 * 发送请求
 * @param $url
 * @param string $postBody
 * @param array $headers
 * @return string
 */
function curlPush($url, $postBody = '', $headers = [])
{
    $curl = curl_init();
    curl_setopt_array($curl, array(
        CURLOPT_URL            => $url,
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_ENCODING       => "",
        CURLOPT_MAXREDIRS      => 10,
        CURLOPT_TIMEOUT        => 30,
        CURLOPT_HTTP_VERSION   => CURL_HTTP_VERSION_1_1,
        CURLOPT_CUSTOMREQUEST  => "POST",
        CURLOPT_POSTFIELDS     => $postBody,
        CURLOPT_HTTPHEADER     => $headers,
    ));
    $response = curl_exec($curl) or $response = "cURL Error #:" . curl_error($curl);
    curl_close($curl);
    return $response;
}

/**
 * 组装请求
 * @param $fileName
 * @param $content
 * @return string
 */
function send($fileName, $content)
{
    $headers = [
        'binfile-md5:  ' . md5($content),
        'binfile-auth: 13YehO6WQOVHN3fY',
        'binfile-gzip: false',
        'Content-Type: multipart/form-data; boundary=syjcxguangdong'
    ];

    $sendStr = <<<EOF
--syjcxguangdong
Content-Disposition: form-data; name="binFile"; filename="$fileName"
Content-Type: application/octet-stream
Content-Transfer-Encoding: binary

$content
--syjcxguangdong
Content-Disposition: form-data; name="filename"
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

$fileName

--syjcxguangdong--
EOF;

    return curlPush(URL, $sendStr, $headers);
}

//=========

/**
 * 打印 for 测试
 */
function _log()
{
    static $files = [];
    $bt = debug_backtrace();
    $caller = array_shift($bt);
    if (!isset($files[$caller['file']])) {
        $files[$caller['file']] = file($caller['file']);
    }

    preg_match('#\((.*)\)#i', $files[$caller['file']][$caller['line'] - 1], $params);

    echo "\nline:", $caller['line'], ": ", $params[1], "> ";

    if (func_num_args() > 1) {
        print_r(func_get_args());
    } else {
        $v = func_get_arg(0);
        if (is_array($v) || is_object($v)) {
            var_dump($v);
        } else {
            echo $v;
        }
    }
}